package yredis

import (
	"dgo/goutils/utils"
	"dgo/plg/plgbase"
	"fmt"
	"gitee.com/ymofen/supervalue"
	"time"
)

type PlgRedisTest struct {
	plgbase.PlgBase
	recvMonitor        *utils.MonitorSpeedSizeRec
	pubMonitor         *utils.MonitorSpeedSizeRec
	sub_channel_prefix string
	srcex              *plgbase.InnerEx
	subex              *plgbase.InnerPlug

	pub_channel_prefix string
	pub_channel_num    int
	pub_min_interval   time.Duration
	pubex              *plgbase.InnerPlug

	lastPubT time.Time
}

/*
**

	{
	  "id":"redis-test",
	  "type":"yredis.test",
	  "conf":{
		 "sub-ex":{"data_ex", "channel-prefix":"vrs-", "channel-num":1000},
	     "pub-ex":{"data_ex", "channel-prefix":"vrs-", "channel-num":1000, "src":{"id":"data_ex", "topic":"VRSRAW"}}
	   }
	}
*/
func (this *PlgRedisTest) ReloadConf(conf *wrapper.SuperValue) error {
	if this.recvMonitor == nil {
		this.recvMonitor = utils.NewMonitorSpeedSizeRec()
	}

	if this.srcex == nil {
		this.srcex = plgbase.NewInnerEx()
		this.srcex.ConfigOwnerPlug(this)
	}

	if this.subex == nil {
		this.subex = plgbase.NewInnerPlug2(this)
	}

	if this.pubex == nil {
		this.pubex = plgbase.NewInnerPlug2(this)
	}

	conf.CheckExistsStrFunc("sub-ex.id", func(strv string) {
		this.subex.ConfigInnerPlugId(strv)
	})

	conf.CheckExistsStrFunc("sub-ex.channel-prefix", func(strv string) {
		this.sub_channel_prefix = strv
	})

	conf.CheckExistsIntFunc("sub-ex.channel-num", func(val int64) {
		prefix := this.sub_channel_prefix
		for i := 0; i < int(val); i++ {
			topic := fmt.Sprintf("%s-%d", prefix, i)
			this.subex.SubscribeEx(topic, fmt.Sprintf("%s-%s", this.PlugId, topic), func(sender interface{}, channelid string, subscribeid interface{}, data interface{}, tag interface{}, isBreak *bool) bool {
				return true
			})
		}
	})

	conf.CheckExistsStrFunc("pub-ex.id", func(strv string) {
		this.pubex.ConfigInnerPlugId(strv)
	})

	conf.CheckExistsStrFunc("pub-ex.channel-prefix", func(strv string) {
		this.pub_channel_prefix = strv
	})

	conf.CheckExistsIntFunc("pub-ex.channel-num", func(val int64) {
		this.pub_channel_num = int(val)
	})

	conf.CheckExistsIntFunc("pub-ex.min-interval-secs", func(val int64) {
		this.pub_min_interval = time.Duration(val) * time.Second
	})

	this.srcex.ReloadConfAndSubscribe(conf.ValueByPath("pub-ex.src"), func(sender interface{}, channelid string, subscribeid interface{}, data interface{}, tag interface{}, isBreak *bool) bool {
		if time.Since(this.lastPubT) < this.pub_min_interval {
			return true
		}
		this.lastPubT = time.Now()
		for i := 0; i < this.pub_channel_num; i++ {
			channelid := fmt.Sprintf("%s-%d", this.pub_channel_prefix, i)
			utils.DefaultWorkers().PostTaskFunc(channelid, func() {
				this.pubex.DispatchData(channelid, this, data, tag)
			})
		}

		return true
	})

	return nil
}

func CreatePlgRedisTestFunc(args ...interface{}) (rval interface{}, err error) {
	obj := &PlgRedisTest{}
	if len(args) > 0 {
		if str, ok := args[0].(string); ok {
			obj.PlugId = str
		}
	}
	rval = obj
	return
}
